﻿using System;
using System.Collections.Generic;
using System.Data;
using System.Data.SqlClient;
using System.IO;
using System.Reflection;
using System.Text;
using System.Threading;
using CoreLab.Oracle;
using Common.Utilities;
using Common.Utilities.Dispatcher;

namespace Workflows.Components.ETL
{
    /// <summary>
    /// 
    /// </summary>
    public class FuzzyLinkWorker:IWorker
    {
        #region fields
        private Thread _WorkerThread;
        private FuzzyLinkInstruction _Instruction;
        private FuzzyLinkOutput _Output;
        private Guid _JobTicketID;
        private RunContext _Context;
        private string _WorkflowName;
        private string _ComponentName;

        /// <summary>
        /// 
        /// </summary>
        public string ComponentName
        {
            get { return _ComponentName; }
        }

        /// <summary>
        /// 
        /// </summary>
        public string WorkflowName
        {
            get { return _WorkflowName; }
        }
        #endregion

        #region IWorker Members

        public Thread WorkerThread
        {
            get { return _WorkerThread; }
        }

        public InstructionBase Instruction
        {
            get { return _Instruction; }
        }

        public ResultBase Output
        {
            get { return _Output; }
        }

        public Type ComponentType
        {
            get { return typeof(FuzzyLinkComponent); }
        }

        public RunContext Context
        {
            get { return _Context; }
            set { _Context = value; }
        }

        public void SetInstruction(Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._JobTicketID = ticketGuid;
            this._Instruction = (FuzzyLinkInstruction)request;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }

        public void SetInstruction(RunContext context, Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._JobTicketID = ticketGuid;
            this._Instruction = (FuzzyLinkInstruction)Instruction;
            this._Context = context;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }

        public void DoWork()
        {
            if (this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._JobTicketID, this, this.GetType(), this._Instruction);
            }
            this._WorkerThread = Thread.CurrentThread;
            this.RunFuzzyLink();
        }


        public void DoWorkAsync()
        {
            if (this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._JobTicketID, this, this.GetType(), this._Instruction);
            }
            this._WorkerThread = new Thread(new ThreadStart(this.RunFuzzyLink));
            this._WorkerThread.Start();
        }

        public void CancelWork()
        {
            if (this._WorkerThread != null && this._WorkerThread.IsAlive)
            {
                this._WorkerThread.Abort();
                if (this.WorkerCancelledEvent != null)
                {
                    this.WorkerCancelledEvent(this._JobTicketID, this.GetType(), this._Instruction);
                }
            }

        }

        public event WorkerStarted WorkerStartedEvent;
        public event WorkerFinished WorkerFinishedEvent;
        public event WorkerHasError WorkerErredEvent;
        public event WorkerCancelled WorkerCancelledEvent;
        public event WorkerStatusChanged WorkerStatusChangedEvent;

        #endregion

        #region work
        private void RunFuzzyLink()
        {
            IDbConnection conn1 = null;
            if (this._Instruction.SrcDbProvider1 == DataProviderType.Oracle)
                conn1 = new OracleConnection(this._Instruction.SrcConnStr1);
            else
                conn1 = new SqlConnection(this._Instruction.SrcConnStr1);
            IDbConnection conn2 = null;
            if (this._Instruction.SrcDbProvider2 == DataProviderType.Oracle)
                conn2 = new OracleConnection(this._Instruction.SrcConnStr2);
            else
                conn2 = new SqlConnection(this._Instruction.SrcConnStr2);
            IDbConnection conn3 = null;
            if (this._Instruction.TgtDbProvider == DataProviderType.Oracle)
                conn3 = new OracleConnection(this._Instruction.TgtConnStr);
            else
                conn3 = new SqlConnection(this._Instruction.TgtConnStr);
            try
            {
                string binFolderPath = Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location);
                string assemblyFilePath = Path.Combine(binFolderPath, this._Instruction.ComparerAssemblyName);
                if (!File.Exists(assemblyFilePath))
                    throw new Exception("Unable to find assembly file path:" + assemblyFilePath);

                Assembly assembly = Assembly.LoadFile(assemblyFilePath);
                Type type = assembly.GetType(this._Instruction.ComparerTypeName);
                IFuzzyFieldComparer comparer = (IFuzzyFieldComparer) Activator.CreateInstance(type);
                if (comparer == null)
                    throw new Exception("Unable to crate comparer instance");

                TableColumn[] srcCols1 = DbUtil.GetTableColumns(
                    this._Instruction.SrcDbProvider1,
                    this._Instruction.SrcConnStr1,
                    this._Instruction.SrcSql1);
                TableColumn[] srcCols2 = DbUtil.GetTableColumns(
                    this._Instruction.SrcDbProvider2,
                    this._Instruction.SrcConnStr2,
                    this._Instruction.SrcSql2);
                Dictionary<string, SqlDbType> srcColDbTypes = new Dictionary<string, SqlDbType>();
                foreach(TableColumn col in srcCols2)
                {
                    srcColDbTypes.Add(col.ColumnName, col.DbType);
                }
                List<TableColumn> tgtCols = new List<TableColumn>();
                TableColumn linkScoreCol = new TableColumn();
                linkScoreCol.ColumnName = this._Instruction.LinkScoreFieldInTargetTable;
                linkScoreCol.DbType = SqlDbType.Decimal;
                linkScoreCol.Size = FieldDataType.GetFieldDefaultSize(SqlDbType.Decimal);
                tgtCols.Add(linkScoreCol);
                Dictionary<string, string> tgt_src1_fieldMappings = new Dictionary<string, string>();
                if(this._Instruction.Src1TgtFieldMappings !=null && this._Instruction.Src1TgtFieldMappings.Count>0)
                {
                    foreach(string srcFieldName in this._Instruction.Src1TgtFieldMappings.Keys)
                    {
                        TableColumn tgtCol = new TableColumn();
                        tgtCol.ColumnName = this._Instruction.Src1TgtFieldMappings[srcFieldName];
                        foreach(TableColumn col in srcCols1)
                        {
                            if(col.ColumnName.ToLower()==srcFieldName.ToLower())
                            {
                                tgtCol.DbType = col.DbType;
                                tgtCol.Size = col.Size;
                                break;
                            }
                        }
                        tgtCols.Add(tgtCol);
                        tgt_src1_fieldMappings.Add(tgtCol.ColumnName, srcFieldName);
                    }
                }
                Dictionary<string, string> tgt_Src2_FieldMappings = new Dictionary<string, string>();
                if(this._Instruction.Src2TgtFieldMappings !=null && this._Instruction.Src2TgtFieldMappings.Count>0)
                {
                    foreach(string srcFieldName in this._Instruction.Src2TgtFieldMappings.Keys)
                    {
                        TableColumn tgtCol = new TableColumn();
                        tgtCol.ColumnName = this._Instruction.Src2TgtFieldMappings[srcFieldName];
                        foreach(TableColumn col in srcCols2)
                        {
                            if(col.ColumnName.ToLower()==srcFieldName.ToLower())
                            {
                                tgtCol.DbType = col.DbType;
                                tgtCol.Size = col.Size;
                                break;
                            }
                        }
                        tgtCols.Add(tgtCol);
                        tgt_Src2_FieldMappings.Add(tgtCol.ColumnName, srcFieldName);
                    }
                }
                string spName = "bulkinsert_" + this._Instruction.TgtTableName + "_SP";
                string spBody = StoredProcUtil.CreateStoredProcedureStatementForInsert(
                    this._Instruction.TgtDbProvider, spName, string.Empty, this._Instruction.TgtTableName,
                    tgtCols.ToArray());
                DbUtil.NewStoredProc(this._Instruction.TgtConnStr, spName, spBody);

                Dictionary<string, IDataParameter> parameterCollection = new Dictionary<string, IDataParameter>();
                foreach(TableColumn tgtCol in tgtCols)
                {
                    IDataParameter parameter = null;
                    if (this._Instruction.TgtDbProvider == DataProviderType.MSSQL)
                        parameter = new SqlParameter("@" + tgtCol.ColumnName, DBNull.Value);
                    else
                        parameter = new OracleParameter("@" + tgtCol.ColumnName, DBNull.Value);
                    parameterCollection.Add(tgtCol.ColumnName, parameter);
                }

                int timeOutSeconds = 15*60;
                int srcRecCount1 =
                    DbUtil.GetTableRecordCount(this._Instruction.SrcDbProvider1,
                                               this._Instruction.SrcConnStr1,
                                               this._Instruction.SrcSql1, timeOutSeconds);
                int srcRecCount2 =
                    DbUtil.GetTableRecordCount(this._Instruction.SrcDbProvider2, this._Instruction.SrcConnStr2,
                                               this._Instruction.SrcSql2, timeOutSeconds);
                int linkedRecCount = 0;

                conn1.Open();
                conn2.Open();
                conn3.Open();

                IDbCommand cmd1 = conn1.CreateCommand();
                cmd1.CommandType = CommandType.Text;

                IDbCommand cmd2 = conn2.CreateCommand();
                cmd2.CommandType = CommandType.Text;

                IDbCommand cmd3 = conn3.CreateCommand();
                cmd3.CommandType = CommandType.Text;

                string key1NameInTgt = this._Instruction.Src1TgtFieldMappings[this._Instruction.SrcKeyField1];

                cmd1.CommandText = this._Instruction.SrcSql1;  
                IDataReader reader1 = cmd1.ExecuteReader();
                int recordProcessed = 0;
                while(reader1.Read())
                {
                    bool skip = false;
                    string key1 = reader1[this._Instruction.SrcKeyField1].ToString();
                    if (reader1[this._Instruction.SrcField1] == null)
                        skip = true;

                    if (!skip)
                    {
                        if (this._Instruction.SkipUpdateExistingData)
                        {
                            cmd3.CommandText = "select count (*) from " +
                                               this._Instruction.TgtTableName + " where " +
                                               key1NameInTgt + "=" + key1;
                            try
                            {
                                int keyExistInTgtTable = int.Parse(cmd3.ExecuteScalar().ToString());
                                if (keyExistInTgtTable > 0)
                                    skip = true;
                            }
                            catch
                            {
                                skip = false;
                            }
                        }
                    }


                    if (!skip)
                    {
                        string mappableField1 = reader1[this._Instruction.SrcField1].ToString();
                        string joinSql = string.Empty;
                        if (this._Instruction.Src1_Src2_Joinings != null &&
                            this._Instruction.Src1_Src2_Joinings.Count > 0)
                        {
                            string[] srcKeys1 = new string[this._Instruction.Src1_Src2_Joinings.Count];
                            this._Instruction.Src1_Src2_Joinings.Keys.CopyTo(srcKeys1, 0);
                            for (int i = 0; i < srcKeys1.Length; i++)
                            {
                                string joinKey2 = this._Instruction.Src1_Src2_Joinings[srcKeys1[i]];
                                SqlDbType dbType = srcColDbTypes[joinKey2];
                                if (reader1[srcKeys1[i]] != null)
                                {
                                    if (joinSql.Length > 0)
                                        joinSql += " and ";
                                    if (this._Instruction.SrcDbProvider2 == DataProviderType.MSSQL)
                                        joinSql +=
                                            joinKey2 + "=" +
                                            FieldDataType.ToMssqlDataValue(dbType, reader1[srcKeys1[i]].ToString());
                                    else
                                        joinSql +=
                                            joinKey2 + "=" +
                                            FieldDataType.ToOracleDataValue(dbType, reader1[srcKeys1[i]].ToString());
                                }
                            }
                        }
                        if (joinSql.Length > 0)
                            joinSql = " where " + joinSql;
                        string sql2 = string.Format(
                            "select * from ({0}) temp {1}",
                            this._Instruction.SrcSql2, joinSql);
                        cmd2.CommandText = sql2;
                        IDataReader reader2 = cmd2.ExecuteReader();
                        while (reader2.Read())
                        {
                            string key2 = reader2[this._Instruction.SrcKeyField2].ToString();
                            if (reader2[this._Instruction.SrcField2] == null)
                                continue;

                            string mappableField2 = reader2[this._Instruction.SrcField2].ToString();
                            double similarity = comparer.CompareSimilarity(mappableField1, mappableField2);
                            if (similarity >= this._Instruction.FieldSimilarityScoreThreshold)
                            {
                                bool existInTgt = false;
                                bool foundBetterMatch = false;
                                if (
                                    this._Instruction.Src1TgtFieldMappings.ContainsKey(this._Instruction.SrcKeyField1) &&
                                    this._Instruction.Src2TgtFieldMappings.ContainsKey(this._Instruction.SrcKeyField2))
                                {
                                    cmd3.CommandType = CommandType.Text;
                                    cmd3.CommandText = "select max (" + this._Instruction.LinkScoreFieldInTargetTable
                                                       + ") from " + this._Instruction.TgtTableName + " where " +
                                                       this._Instruction.Src1TgtFieldMappings[
                                                           this._Instruction.SrcKeyField1] +
                                                       "=" + key1 + " and " +
                                                       this._Instruction.Src2TgtFieldMappings[
                                                           this._Instruction.SrcKeyField2] +
                                                       "=" + key2;
                                    try
                                    {
                                        double linkScore = double.Parse(cmd3.ExecuteScalar().ToString());
                                        if (linkScore >= this._Instruction.FieldSimilarityScoreThreshold)
                                            existInTgt = true;
                                        if (linkScore < similarity)
                                        {
                                            foundBetterMatch = true;
                                            cmd3.CommandText = "delete from " + this._Instruction.TgtTableName +
                                                               " where " +
                                                               this._Instruction.Src1TgtFieldMappings[
                                                                   this._Instruction.SrcKeyField1] +
                                                               "=" + key1 + " and " +
                                                               this._Instruction.Src2TgtFieldMappings[
                                                                   this._Instruction.SrcKeyField2] +
                                                               "=" + key2;
                                            cmd3.ExecuteNonQuery();
                                        }
                                    }
                                    catch
                                    {
                                        existInTgt = false;
                                    }
                                }
                                if (!existInTgt || foundBetterMatch)
                                {
                                    cmd3.CommandType = CommandType.StoredProcedure;
                                    cmd3.CommandText = spName;
                                    cmd3.Parameters.Clear();
                                    foreach (string tgtColName in parameterCollection.Keys)
                                    {
                                        IDataParameter parameter = parameterCollection[tgtColName];
                                        if (tgtColName == this._Instruction.LinkScoreFieldInTargetTable)
                                            parameter.Value = similarity;
                                        else if (tgt_src1_fieldMappings.ContainsKey(tgtColName))
                                            parameter.Value = reader1[tgt_src1_fieldMappings[tgtColName]];
                                        else if (tgt_Src2_FieldMappings.ContainsKey(tgtColName))
                                            parameter.Value = reader2[tgt_Src2_FieldMappings[tgtColName]];
                                        else
                                            parameter.Value = DBNull.Value;
                                        cmd3.Parameters.Add(parameter);
                                    }
                                    cmd3.ExecuteNonQuery();

                                    if (!existInTgt)
                                        linkedRecCount++;
                                }
                            }
                        }
                        reader2.Close();
                    }
                    recordProcessed++;

                    if(this.WorkerStatusChangedEvent !=null && recordProcessed % 100==0)
                    {
                        string msg = string.Format(
                            "Processed {0} of {1} from src1, {2} record linked", 
                            recordProcessed, srcRecCount1, linkedRecCount);
                        int pct = recordProcessed * 100 / srcRecCount1;
                        this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                            this.WorkflowName, this.ComponentName, msg, pct);
                    }
                }
                reader1.Close();

                this._Output = new FuzzyLinkOutput(
                    this._Instruction.OwnerComponentID, this._JobTicketID, srcRecCount1,
                    srcRecCount2, linkedRecCount);
                
                if(this.WorkerFinishedEvent !=null)
                {
                    this.WorkerFinishedEvent(this._JobTicketID, this.GetType(), this._Instruction, this._Output);
                }
            }
            catch(Exception ex)
            {
                if (this.WorkerErredEvent != null)
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(), true);
            }
            finally
            {
                conn1.Close();
                conn2.Close();
                conn3.Close();
            }
        }
        #endregion
    }
}
